package com.kafka_test.app;

import com.kafka_test.app.service.OffsetService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Optional;

@Component
public class GetKafkaData {
    private final OffsetService offsetService;

    @Autowired
    public GetKafkaData(OffsetService offsetService) {
        this.offsetService = offsetService;
    }

    @KafkaListener(topics = {"producer_topic"})
    public void listener(ConsumerRecord<?,?> record){
        Optional<?> msg = Optional.ofNullable(record.value());
        if(msg.isPresent()){
            String str = msg.get().toString().replace("=", ":");
            offsetService.insertConsumer("consumer_3", str);
            System.out.println("listener : "+msg.get());
        }
    }

}
